package alfred.camel;

import alfred.mdp.client.Client;
import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.impl.DefaultProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZContext;
import org.zeromq.ZDispatcher;
import org.zeromq.ZMsg;

import java.util.concurrent.TimeUnit;

/**
 */
public class AlfredProducer extends DefaultProducer {

    private static final Logger LOG = LoggerFactory.getLogger(AlfredProducer.class);
    private final Client client;
    private final String serviceName;
    private final TimeUnit timeUnit;
    private final long timeout;

    public AlfredProducer(AlfredEndpoint endpoint, ZContext zmqContext, ZDispatcher dispatcher, String serviceName, TimeUnit timeUnit, long timeout) {
        super(endpoint);
        this.serviceName = serviceName;
        this.timeUnit = timeUnit;
        this.timeout = timeout;
        client = new Client(zmqContext, dispatcher, endpoint.getComponent().getClientsUri());
    }

    @Override
    public void process(Exchange exchange) throws Exception {
        LOG.debug("Sending message with Alfred client.");
        try {
            ZMsg body = exchange.getIn().getBody(ZMsg.class);
            if (body == null) {
                LOG.warn("Exchange body must be not null, Exchange:{}", exchange);

            } else {
                ZMsg response = client.request(serviceName, body, timeUnit, timeout);
                exchange.getOut().setBody(response);
            }

        } catch (Exception ex) {
            LOG.error("Error occurred while sending message with Alfred client.", ex);
            throw new RuntimeCamelException(ex);

        } finally {
            LOG.debug("Message sent with Alfred client.");

        }
    }
}
